package cn.dmp.report

import cn.dmp.util.AdApi
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext

/**
  * Created by Administrator on 2018/4/22.
  */
object TerritoryAnalyzeCore {

  def main(args: Array[String]): Unit = {

    //判断路径参数
    if (args.length!=2){
      println(
        """
          |inputDataPath,outputDataPath
          |参数路径不合法
        """.stripMargin)
      sys.exit()
    }
    //创建sparkContext
    val conf = new SparkConf()
      .setMaster("local[*]").setAppName("TerritoryAnalyze")
      .set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
    val sc = new SparkContext(conf)

    //输入 输出的路径参数
    val Array(inputDataPath,outPutPath)=args

    //读parquent文件要用SQLContext
    val sqlContext: SQLContext = new SQLContext(sc)
    val dataFrame = sqlContext.read.parquet(inputDataPath)

    dataFrame.map(AdApi(_)).reduceByKey(
      (t1,t2)=>t1.zip(t2).map(f=>f._1+f._2)
    ).map({
      t=>t._1._1+","+t._1._2+","+t._2.mkString(",")
    }).saveAsTextFile(outPutPath)
    sc.stop()
  }
}
